package com.xzx.flume;

import com.fasterxml.jackson.databind.JsonNode;
import lombok.extern.slf4j.Slf4j;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2025/5/13 15:30
 */
@Slf4j
public class TableNameAndTimestampInterceptor implements Interceptor {

    @Override
    public void initialize() {
        //服务启动的时候掉一次
        log.info("custom plugin init......");
    }

    @Override
    public Event intercept(Event event) {
        String body = new String(event.getBody(), StandardCharsets.UTF_8);
        JsonNode json = JsonUtil.parseJson(body);
        String database = JsonUtil.asText(json, "database");
        String table = JsonUtil.asText(json, "table");
        // 将数据量的毫秒转成秒
        long ts = JsonUtil.asLong(json, "ts") * 1000;
        Map<String, String> headers = event.getHeaders();
        headers.put("tableName", table);
        headers.put("database", database);
        headers.put("timestamp", String.valueOf(ts));
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        if (Objects.nonNull(list)) {
            for (Event event : list) {
                intercept(event);
            }
        }
        return list;
    }

    @Override
    public void close() {
        //服务关闭的时候掉一次
        log.info("custom plugin close......");
    }

     public static class Builder implements org.apache.flume.interceptor.Interceptor.Builder{
        //获取自定义拦截器对象
        @Override
        public Interceptor build() {
            return new TableNameAndTimestampInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
